package com.mr;

import java.io.IOException;
import java.util.Arrays;
import java.util.Date;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ClearBookMR {

  public static class ClearBookMapper extends Mapper<Text, Text, Text, Text> {

    @Override
    protected void map(Text key, Text value, Context context)
        throws IOException, InterruptedException {
      String line = key.toString();

      if (StringUtils.isNotEmpty(line)) {
        //不需要替换" 替换&amp;
        String[] arr = line.replaceAll("&amp;", "").split(";");
        String[] newArr = Arrays.copyOf(arr, arr.length + 1);
        String md5 = MD5Encrypt.encode(arr[0].replaceAll("\"", ""));
        newArr[arr.length] = "\"" + md5 + "\"";
        context.write(new Text(String.join(";", newArr)), new Text(""));

      }
    }
  }

  public static class ClearBookReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
        throws IOException, InterruptedException {
      for (Text value : values) {
        context.write(key, null);
      }
    }
  }

  public static void main(String[] args)
      throws InterruptedException, IOException, ClassNotFoundException {
    Configuration conf = new Configuration();
    org.apache.hadoop.mapreduce.Job job = org.apache.hadoop.mapreduce.Job.getInstance(conf);

    job.setJarByClass(ClearBookMR.class);

    job.setInputFormatClass(KeyValueTextInputFormat.class);

    job.setMapperClass(ClearBookMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    FileInputFormat.setInputPaths(job, new Path(args[0]));

    job.setReducerClass(ClearBookReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileOutputFormat.setOutputPath(job, new Path(args[1] + "/" + new Date().getTime()));

    job.waitForCompletion(true);
  }
}
